package com.example.cuckoofilterdemo.filter;

import com.google.common.hash.Hashing;

import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Component
public class RedisCuckooFilter {
    private final StringRedisTemplate redisTemplate;
    private final int maxKicks;
    private final int fingerprintSize;

    private static final String CUCKOO_FILTER_KEY = "cuckoo:filter";
    private static final String CUCKOO_FILTER_LOCK_KEY = "cuckoo:filter:lock";

    public RedisCuckooFilter(StringRedisTemplate redisTemplate,
                             int maxKicks,
                             int fingerprintSize) {
        this.redisTemplate = redisTemplate;
        this.maxKicks = maxKicks;
        this.fingerprintSize = fingerprintSize;
    }

    private long[] getBuckets(String item) {
        byte[] hash = Hashing.murmur3_128().hashString(item, StandardCharsets.UTF_8).asBytes();
        long h1 = getLongFromBytes(hash, 0);
        long h2 = getLongFromBytes(hash, 8);
        return new long[]{h1, h2};
    }

    private String getFingerprint(String item) {
        byte[] hash = Hashing.murmur3_32().hashString(item, StandardCharsets.UTF_8).asBytes();
        byte[] fingerprint = Arrays.copyOfRange(hash, 0, fingerprintSize);
        return new String(fingerprint, StandardCharsets.UTF_8);
    }

    private long getLongFromBytes(byte[] bytes, int offset) {
        long value = 0;
        for (int i = 0; i < 8; i++) {
            value = (value << 8) | (bytes[offset + i] & 0xff);
        }
        return value;
    }

    public boolean mightContain(String item) {
        String fingerprint = getFingerprint(item);
        long[] buckets = getBuckets(item);

        String bucket1 = (String) redisTemplate.opsForHash().get(CUCKOO_FILTER_KEY, Arrays.toString(buckets));
        String bucket2 = (String) redisTemplate.opsForHash().get(CUCKOO_FILTER_KEY, String.valueOf(buckets[1]));


        return (bucket1 != null && bucket1.contains(fingerprint)) ||
                (bucket2 != null && bucket2.contains(fingerprint));
    }

    public boolean add(String item) {
        String fingerprint = getFingerprint(item);
        long[] buckets = getBuckets(item);

        // 尝试获取锁
        Boolean locked = redisTemplate.opsForValue().setIfAbsent(
                CUCKOO_FILTER_LOCK_KEY,
                "1",
                10,
                TimeUnit.SECONDS
        );

        if (locked == null || !locked) {
            return false;
        }

        try {
            for (int i = 0; i < maxKicks; i++) {
                for (long bucket : buckets) {
                    String bucketKey = String.valueOf(bucket);
                    String current = (String) redisTemplate.opsForHash().get(CUCKOO_FILTER_KEY, bucketKey);

                    if (current == null || current.length() < fingerprintSize * 10) {
                        String newValue = (current == null) ? fingerprint : current + fingerprint;
                        redisTemplate.opsForHash().put(CUCKOO_FILTER_KEY, bucketKey, newValue);
                        return true;
                    }
                }

                // 随机选择一个桶进行踢出操作
                long bucket = buckets[i % 2];
                String bucketKey = String.valueOf(bucket);
                String current = (String) redisTemplate.opsForHash().get(CUCKOO_FILTER_KEY, bucketKey);

                if (current != null && current.length() >= fingerprintSize) {
                    // 踢出第一个指纹
                    String victim = current.substring(0, fingerprintSize);
                    String remaining = current.substring(fingerprintSize);
                    redisTemplate.opsForHash().put(CUCKOO_FILTER_KEY, bucketKey, remaining);

                    // 重新插入被踢出的指纹
                    fingerprint = victim;
                    buckets = getBuckets(victim);
                }
            }
        } finally {
            // 释放锁
            redisTemplate.delete(CUCKOO_FILTER_LOCK_KEY);
        }

        return false;
    }
}
